#include "messaging/kafka_consumer.h"
#include "common/logger/logger.h"
#include <librdkafka/rdkafkacpp.h>

using namespace common::logger;

namespace messaging {

// 实现消费者回调
class ConsumerCb : public RdKafka::ConsumeCb {
public:
    explicit ConsumerCb(MessageCallback callback) : callback_(callback) {}

    void consume_cb(RdKafka::Message& message, void* /* opaque */) override {
        if (message.err() == RdKafka::ERR_NO_ERROR) {
            std::string key;
            std::string value;
            
            if (message.key()) {
                key = std::string(reinterpret_cast<const char*>(message.key()), message.key_len());
            }
            
            if (message.payload()) {
                value = std::string(reinterpret_cast<const char*>(message.payload()), message.len());
            }
            
            callback_(message.topic_name(), key, value, message.offset());
        } else if (message.err() != RdKafka::ERR__PARTITION_EOF) {
            LOG_ERROR("Kafka消费错误: " + message.errstr());
        }
    }

private:
    MessageCallback callback_;
};

// 实现错误回调
class ConsumerErrorCb : public RdKafka::EventCb {
public:
    explicit ConsumerErrorCb(ErrorCallback callback) : callback_(callback) {}

    void event_cb(RdKafka::Event& event) override {
        switch (event.type()) {
            case RdKafka::Event::EVENT_ERROR:
                LOG_ERROR("Kafka错误: " + event.str());
                if (callback_) {
                    callback_(event.str());
                }
                break;
            case RdKafka::Event::EVENT_STATS:
                LOG_INFO("Kafka统计: " + event.str());
                break;
            case RdKafka::Event::EVENT_LOG:
                LOG_INFO("Kafka日志: " + event.str());
                break;
            default:
                LOG_INFO("Kafka事件: " + event.str());
                break;
        }
    }

private:
    ErrorCallback callback_;
};

// 创建Kafka消费者
std::shared_ptr<KafkaConsumer> KafkaConsumer::create(const KafkaConsumerConfig& config) {
    auto consumer = std::shared_ptr<KafkaConsumer>(new KafkaConsumer(config));
    if (!consumer->init(config)) {
        return nullptr;
    }
    return consumer;
}

KafkaConsumer::KafkaConsumer(const KafkaConsumerConfig& /* config */) 
    : isRunning_(false) {
}

KafkaConsumer::~KafkaConsumer() {
    stop();
    if (consumer_) {
        consumer_->close();
        consumer_.reset();
    }
}

bool KafkaConsumer::init(const KafkaConsumerConfig& config) {
    try {
        // 创建配置对象
        std::string errstr;
        RdKafka::Conf* conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        
        // 设置基本配置
        conf->set("bootstrap.servers", config.brokers, errstr);
        conf->set("group.id", config.groupId, errstr);
        conf->set("client.id", config.clientId, errstr);
        conf->set("session.timeout.ms", std::to_string(config.sessionTimeoutMs), errstr);
        conf->set("max.poll.interval.ms", std::to_string(config.maxPollIntervalMs), errstr);
        conf->set("auto.offset.reset", config.offsetReset, errstr);
        conf->set("fetch.message.max.bytes", std::to_string(config.fetchMaxBytes), errstr);
        conf->set("fetch.wait.max.ms", std::to_string(config.fetchMaxWaitMs), errstr);
        conf->set("enable.auto.commit", "false", errstr);  // 手动提交偏移量
        
        // 创建消费者
        RdKafka::KafkaConsumer* consumer = RdKafka::KafkaConsumer::create(conf, errstr);
        if (!consumer) {
            LOG_ERROR("创建Kafka消费者失败: " + errstr);
            delete conf; // 如果创建失败，需要手动删除配置对象
            return false;
        }
        
        // 转移所有权
        conf_.reset(conf);
        consumer_.reset(consumer);
        
        LOG_INFO("Kafka消费者已创建，brokers: " + config.brokers + ", group: " + config.groupId);
        return true;
    } catch (const std::exception& e) {
        LOG_ERROR("初始化Kafka消费者异常: " + std::string(e.what()));
        return false;
    }
}

bool KafkaConsumer::subscribe(const std::vector<std::string>& topics) {
    if (!consumer_) {
        LOG_ERROR("Kafka消费者未初始化");
        return false;
    }
    
    // 订阅主题
    RdKafka::ErrorCode err = consumer_->subscribe(topics);
    if (err != RdKafka::ERR_NO_ERROR) {
        LOG_ERROR("订阅主题失败: " + RdKafka::err2str(err));
        return false;
    }
    
    LOG_INFO("已订阅主题: " + std::to_string(topics.size()) + "个");
    return true;
}

void KafkaConsumer::unsubscribe() {
    if (consumer_) {
        consumer_->unsubscribe();
        LOG_INFO("已取消订阅所有主题");
    }
}

bool KafkaConsumer::start(MessageCallback messageCallback, ErrorCallback errorCallback) {
    if (!consumer_) {
        LOG_ERROR("Kafka消费者未初始化");
        return false;
    }
    
    if (isRunning_) {
        LOG_WARNING("Kafka消费者已经在运行");
        return true;
    }
    
    // 保存回调
    messageCallback_ = messageCallback;
    errorCallback_ = errorCallback;
    
    // 设置错误回调
    if (errorCallback) {
        std::string errstr;
        auto errorCb = new ConsumerErrorCb(errorCallback);
        conf_->set("event_cb", errorCb, errstr);
    }
    
    // 启动消费线程
    isRunning_ = true;
    consumeThread_ = std::thread(&KafkaConsumer::consumeThread, this);
    
    LOG_INFO("Kafka消费者已启动");
    return true;
}

void KafkaConsumer::stop() {
    if (!isRunning_) {
        return;
    }
    
    // 停止消费线程
    isRunning_ = false;
    condition_.notify_all();
    
    if (consumeThread_.joinable()) {
        consumeThread_.join();
    }
    
    LOG_INFO("Kafka消费者已停止");
}

bool KafkaConsumer::commitSync() {
    if (!consumer_) {
        LOG_ERROR("Kafka消费者未初始化");
        return false;
    }
    
    RdKafka::ErrorCode err = consumer_->commitSync();
    if (err != RdKafka::ERR_NO_ERROR) {
        LOG_ERROR("提交偏移量失败: " + RdKafka::err2str(err));
        return false;
    }
    
    return true;
}

void KafkaConsumer::commitAsync() {
    if (consumer_) {
        consumer_->commitAsync();
    }
}

bool KafkaConsumer::seek(const std::string& topic, int partition, int64_t offset) {
    if (!consumer_) {
        LOG_ERROR("Kafka消费者未初始化");
        return false;
    }
    
    // 创建分区对象
    std::vector<RdKafka::TopicPartition*> partitions;
    RdKafka::TopicPartition* tp = RdKafka::TopicPartition::create(topic, partition, offset);
    partitions.push_back(tp);
    
    // 设置偏移量
    RdKafka::ErrorCode err = consumer_->assign(partitions);
    
    // 释放资源
    for (auto& p : partitions) {
        delete p;
    }
    
    if (err != RdKafka::ERR_NO_ERROR) {
        LOG_ERROR("设置偏移量失败: " + RdKafka::err2str(err));
        return false;
    }
    
    return true;
}

void KafkaConsumer::consumeThread() {
    while (isRunning_) {
        // 消费消息
        RdKafka::Message* message = consumer_->consume(100); // 超时时间100ms
        
        if (message) {
            // 处理消息
            if (message->err() == RdKafka::ERR_NO_ERROR) {
                if (messageCallback_) {
                    std::string key;
                    std::string value;
                    
                    if (message->key()) {
                        key = std::string(reinterpret_cast<const char*>(message->key()), message->key_len());
                    }
                    
                    if (message->payload()) {
                        value = std::string(reinterpret_cast<const char*>(message->payload()), message->len());
                    }
                    
                    messageCallback_(message->topic_name(), key, value, message->offset());
                }
                
                // 自动提交偏移量
                commitAsync();
            } else if (message->err() != RdKafka::ERR__PARTITION_EOF &&
                      message->err() != RdKafka::ERR__TIMED_OUT) {
                LOG_ERROR("Kafka消费错误: " + message->errstr());
                if (errorCallback_) {
                    errorCallback_(message->errstr());
                }
            }
            
            // 释放消息
            delete message;
        }
        
        // 检查是否需要等待
        if (!isRunning_) {
            std::unique_lock<std::mutex> lock(mutex_);
            condition_.wait_for(lock, std::chrono::milliseconds(100));
        }
    }
}

} // namespace messaging 